package com.hyq.learn.rabbitmq.handle;

import com.google.common.base.Stopwatch;
import com.hyq.learn.common.util.SnakeJsonUtils;
import com.hyq.learn.rabbitmq.message.BaseMessage;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.function.Consumer;

/**
 * @author heyuqin
 * @date 2024/8/26 14:47
 */
@Component
public class RabbitMessageHandle {
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMessageHandle.class);

    private static final long DEFAULT_TIMEOUT_MS = 1000;

    /**
     * 消息处理器
     *
     * @param message
     * @param channel
     */
    public <T extends BaseMessage> void handle(Message message, Channel channel, Class<T> clazz, Consumer<T> consumer) {
        Stopwatch stopwatch = Stopwatch.createStarted();
        String json = new String(message.getBody());
        MessageProperties messageProperties = message.getMessageProperties();
        String consumerQueue = messageProperties.getConsumerQueue();
        String receivedExchange = messageProperties.getReceivedExchange();
        String receivedRoutingKey = messageProperties.getReceivedRoutingKey();
        LOGGER.debug("exchange:[{}], routing key: [{}], queue:[{}] message content: {}", receivedExchange, receivedRoutingKey, consumerQueue, json);
        boolean requeue = false;
        try {
            T t = SnakeJsonUtils.fromJson(json, clazz);
            consumer.accept(t);
        } catch (Throwable e) {
            LOGGER.error("handle rabbitmq message error: {}, queue {}, exchange {}, routing key {}, content: {}",
                    e.getMessage(), consumerQueue, receivedExchange, receivedRoutingKey, json, e);
            if (isRequeue(e)) {
                requeue = true;
            }
        } finally {
            long elapsedTimes = stopwatch.elapsed().toMillis();
            if (elapsedTimes >= DEFAULT_TIMEOUT_MS) {
                LOGGER.warn("exchange:[{}], routing key: [{}], queue:[{}] elapsed time {}ms", receivedExchange,
                        receivedRoutingKey, consumerQueue, elapsedTimes);
            } else {
                LOGGER.debug("exchange:[{}], routing key: [{}], queue:[{}] elapsed time {}ms", receivedExchange,
                        receivedRoutingKey, consumerQueue, elapsedTimes);
            }
            try {
                if (requeue) {
                    channel.basicNack(messageProperties.getDeliveryTag(), false, true);
                } else {
                    channel.basicAck(messageProperties.getDeliveryTag(), false);
                }
            } catch (IOException e) {
                LOGGER.error("error to mq ack, cause: {}, content: {}", e.getMessage(), json, e);
            }
        }
    }

    private boolean isRequeue(Throwable e) {
        // 判断是否需要重新入队的方法
        return false;
    }

}
